package com.nami.gate.middleware.mq.listener;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;

import com.nami.gate.cache.LocalChannelCache;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.im.network.model.packet.v2.Packet;
import com.nami.transport.proto.codec.MqTransMsg;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.nami.gate.external.basic.constants.CommonConst.DELTA;


public class DownMsgFlushOneConsumer extends AbstractConsumer {

    private static final Logger log = LoggerFactory.getLogger(DownMsgFlushOneConsumer.class);

    @Override
    public void onMessage(byte[] body) {
        try {
            TimeInterval timer = DateUtil.timer();

            MqTransMsg.Dispatch2GatePublishResponse response = MqTransMsg.Dispatch2GatePublishResponse.parseFrom(body);
            MqTransMsg.MessageHeader msgHeader = response.getMsgHeader();
            if ((System.currentTimeMillis() - response.getTimestamp()) > DELTA) {
                log.warn("[flushOne] 过期信息不予消费。sessionId={}，messageId={}", response.getSessionId(), response.getMsgId());
                return;
            }
            Channel channel = LocalChannelCache.get(response.getSessionId());
            if (channel == null) {
                // 如果为空，暂时为删除
                // TODO 不能这样干好像。排查问题不好排查
//                LocalChannelCache.removeAll(request.getUserInstanceId());
                log.error("[flushOne] 出问题了，通道不见了。sessionId={}", response.getSessionId());
                return;
            }

            Packet packet = new Packet();
            packet.setCmd(msgHeader.getCmd().toByteArray()[0]);
            packet.setQos(msgHeader.getQos().toByteArray()[0]);
            packet.setVersion(msgHeader.getVersion().toByteArray()[0]);
            packet.setBody(response.getDownMsg().toByteArray());
            channel.writeAndFlush(packet);
            if (timer.interval() > 500) {
                log.warn("[flushOne]【{}】发送消息成功 use:{}ms", response.getSessionId(), timer.interval());
            }
        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("[flushOne] 消费出现问题:{}", var3.getMessage());
        }
    }

    @Override
    public String getTopicName() {
        return DownTopicConst.FLUSH_ONE_TOPIC;
    }

}


